package com.wuwangfu.func;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.wuwangfu.entity.LogBean;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Supplier;

/**
 * @Description：异步 IO
 * @Author：jcshen
 * @Date：2023-06-29
 *
 * https://nightlies.apache.org/flink/flink-docs-release-1.17/zh/docs/dev/datastream/operators/asyncio/
 *
 */
public class AsyncHttpGeoFunc extends RichAsyncFunction<String, LogBean> {

    //异步请求的HttpClient
    private transient CloseableHttpAsyncClient httpClient;
    //请求高德地图URL
    private String url;
    //请求高德地图密钥
    private String key;
    //异步HttpClient支持最大连接
    private int maxConnTotal;

    public AsyncHttpGeoFunc(String url, String key, int maxConnTotal) {
        this.url = url;
        this.key = key;
        this.maxConnTotal = maxConnTotal;
    }

    @Override
    public void open(Configuration parameters) throws Exception {

        RequestConfig requestConfig = RequestConfig.custom().build();
        //创建HttpAsyncClient请求连接池
        HttpAsyncClients.custom()
                //设置最大连接数
                .setMaxConnTotal(maxConnTotal)
                .setDefaultRequestConfig(requestConfig).build();
        //启动异步请求 httpClient
        httpClient.start();
    }

    @Override
    public void asyncInvoke(String line, ResultFuture<LogBean> resultFuture) throws Exception {
        //将字符串解析为对象
        LogBean bean = JSON.parseObject(line, LogBean.class);
        //获取经度
        Double longitude = bean.getLongitude();
        //获取维度
        Double latitude = bean.getLatitude();
        //将经纬度和高德地图的key与请求的url进行拼接
        HttpGet httpGet = new HttpGet(url + "?location=" + longitude + "," + latitude + "&key=" + key);
        //发送异步请求，返回Future
        Future<HttpResponse> future = httpClient.execute(httpGet, null);
        CompletableFuture.supplyAsync(new Supplier<LogBean>() {
            @Override
            public LogBean get() {
                try {
                    HttpResponse response = future.get();
                    String province = null;
                    String city = null;
                    if (response.getStatusLine().getStatusCode() == 200){
                        //解析返回的结果，获取省份、城市等信息
                        String result = EntityUtils.toString(response.getEntity());
                        JSONObject json = JSON.parseObject(result);
                        JSONObject regeocode = json.getJSONObject("regeocode");
                        if (regeocode !=null && !regeocode.isEmpty()){
                            JSONObject address = regeocode.getJSONObject("addressComponent");
                            province = address.getString("province");
                            city = address.getString("city");
                        }
                    }
                    //将返回的结果进行赋值
                    bean.setProvince(province);
                    bean.setCity(city);

                    return bean;
                } catch (Exception e) {
                    return null;
                }
            }
        }).thenAccept((LogBean result)->{
            //将结果添加到resultFuture中输出（complete方法的参数只能为集合，如果只有一个元素，就返回一个单例集合）
            resultFuture.complete(Collections.singleton(result));
        });
    }


    @Override
    public void close() throws Exception {
        //关闭HttpAsyncClients请求连接池
        httpClient.close();
    }
}
